package cn.flkj.evidence.RocketMQConsumer.Utils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.elasticsearch.ElasticsearchStatusException;
import org.elasticsearch.action.ActionFuture;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.elasticsearch.search.sort.SortOrder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

public class EsUtil {

    private static Logger log =  LoggerFactory.getLogger(EsUtil.class);

    private static UpdateRequest buidRequst(String index, String id, Map map){

    //    UpdateRequest request = new UpdateRequest (index, "doc", id);
        UpdateRequest request = new UpdateRequest(index,id);
        String[] includes = Strings.EMPTY_ARRAY;
        String[] excludes = new String[]{"updated"};

        request.doc(map);
        request.fetchSource(new FetchSourceContext(true, includes, excludes));
        request.detectNoop(true);
        String jsonString = "{\"test\":\"syslog\"}";
        request.upsert(jsonString, XContentType.JSON);
        request.scriptedUpsert(true);
        request.docAsUpsert(true);

        return request;
    }

    private static UpdateRequest buidRequst(String index, String id, JSONObject obj){

     //   UpdateRequest request = new UpdateRequest (index, "doc", id);
        UpdateRequest request = new UpdateRequest(index,id);
        String[] includes = Strings.EMPTY_ARRAY;
        String[] excludes = new String[]{"updated"};

        request.doc(obj);
        request.fetchSource(new FetchSourceContext(true, includes, excludes));
        request.detectNoop(true);
        String jsonString = "{\"test\":\"syslog\"}";
        request.upsert(jsonString, XContentType.JSON);
        request.scriptedUpsert(true);
        request.docAsUpsert(true);

        return request;
    }

    private static BulkResponse bulk(RestHighLevelClient client,BulkRequest bulkRequest){

        BulkResponse bulkResponse = null;
        try {
            bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error("bulk error:" + e.getMessage());
        }finally {
            return bulkResponse;
        }
    }

    public static void SavePoint(RestHighLevelClient client,String index,String id,Map<String,String> map){
        UpdateRequest request = buidRequst(index,id,map);

        try {
            client.update(request,RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error("SavePoint error:" + index + ":" + id);
        }
    }

    public static String SearchPoint(RestHighLevelClient client,String index,String id){
     //   GetRequest getRequest = new GetRequest(index,"doc",id);
        GetRequest getRequest = new GetRequest(index,id);
        String point = null;
        try {
            GetResponse getResponse = client.get(getRequest,RequestOptions.DEFAULT);
            Map map = getResponse.getSource();
            point = map.get("point").toString();

        } catch (IOException e) {
            log.error("searchPoint error:" + index + ":" + id);
        }finally {
            return point;
        }
    }

    public static boolean Query(RestHighLevelClient client, String index,String id){

       // GetRequest getRequest = new GetRequest(index,"doc",id);
        GetRequest getRequest = new GetRequest(index,id);
        boolean flag = false;

        try {
            GetResponse getResponse = client.get(getRequest,RequestOptions.DEFAULT);

            Map map = getResponse.getSource();

            if(map.containsKey("ledger")){
                String ledger = map.get("ledger").toString();
                if(ledger.length()>0) {
                    flag = true;
                }
            }
        } catch (IOException e) {
            log.error("ES Query error:" + e.getMessage());
        }finally {
            return flag;
        }
    }

    public static void buidIndex(RestHighLevelClient client,String index){

        try {
            GetIndexRequest indexRequest = new GetIndexRequest().indices(index);
            boolean exists = client.indices().exists(indexRequest, RequestOptions.DEFAULT);

            if(!exists) {
                CreateIndexRequest createIndexRequest = new CreateIndexRequest(index);
                XContentBuilder jsonMapping = XContentFactory.jsonBuilder();
                jsonMapping.startObject().startObject("_doc").startObject("properties")
                        .startObject("slsj").field("type", "date").field("format", "yyyy-MM-dd HH:mm:ss").endObject()
                        .endObject().endObject().endObject();
                createIndexRequest.mapping("_doc", jsonMapping);
//                createIndexRequest.mapping("doc", jsonMapping);
                try {
                    CreateIndexResponse response = client.indices().create(createIndexRequest, RequestOptions.DEFAULT);
                } catch (ElasticsearchStatusException e) {
                    //状态码是400 badRequest 所以输出Message
                    log.error("StatusCode:" + e.status().getStatus());
                    log.error("Message:" + e.getMessage());
                }
            }
        } catch (IOException e) {
            log.info("build index error:"+e.getMessage());
        }
    }

    public static void BulkFromMap(RestHighLevelClient client, Map<String, JSONObject> map, String index){
        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//设置日期格式
        String nowTime = df.format(new Date());

        BulkRequest bulkRequest = new BulkRequest();
        for(String id : map.keySet()){
            JSONObject obj = map.get(id);
            if(!obj.containsKey("slsj")){
                obj.put("slsj",nowTime);
            }
         /*
            Map data = JSON.parseObject(obj.toJSONString());
            if( !data.containsKey("slsj")){
                data.put("slsj",nowTime);
            }*/

            bulkRequest.add(buidRequst(index,id,obj));
        }

        BulkResponse bulkResponse = bulk(client,bulkRequest);

        if(bulkResponse != null && bulkResponse.hasFailures()){
            log.info(bulkResponse.buildFailureMessage());

            for (BulkItemResponse bulkItemResponse : bulkResponse) {
                if (bulkItemResponse.isFailed()) {
                    BulkItemResponse.Failure failure = bulkItemResponse.getFailure();
                    boolean flag = Query(client,index,failure.getId());
                    if(!flag) {
                        log.info("EsLog:" + failure.getMessage());
                    }
                }
            }
        }
    }

    public static void BulTxid(RestHighLevelClient client, Set<String> set,Map<String,Integer>filterMap, String index, String ledger, String txid){

        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//设置日期格式
        String nowTime = df.format(new Date());

        BulkRequest bulkRequest = new BulkRequest();
        for(String id : set){
            Map map = new HashMap();
            map.put("filter",filterMap.get(id));
            map.put("slsj",nowTime);
            map.put("txid",txid);
            map.put("ledger",ledger);

            bulkRequest.add(buidRequst(index,id,map));
        }

        BulkResponse bulkResponse = bulk(client,bulkRequest);

        if(bulkResponse != null && bulkResponse.hasFailures()){
            for (BulkItemResponse bulkItemResponse : bulkResponse) {
                if (bulkItemResponse.isFailed()) {
                    BulkItemResponse.Failure failure = bulkItemResponse.getFailure();

                    boolean flag = Query(client,index,failure.getId());
                    if(!flag) {
                        log.info("Txid:" + failure.getMessage());
                    }
                }
            }
        }
    }

    public static void BulTxid(RestHighLevelClient client, Set<String> set, String index, String ledger, String txid){

        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//设置日期格式
        String nowTime = df.format(new Date());

        BulkRequest bulkRequest = new BulkRequest();
        for(String id : set){
            Map map = new HashMap();
            map.put("slsj", nowTime);
            map.put("txid", txid);
            map.put("ledger", ledger);

            bulkRequest.add(buidRequst(index,id,map));
        }

        BulkResponse bulkResponse = bulk(client,bulkRequest);

        if(bulkResponse != null && bulkResponse.hasFailures()){
            for (BulkItemResponse bulkItemResponse : bulkResponse) {
                if (bulkItemResponse.isFailed()) {
                    BulkItemResponse.Failure failure = bulkItemResponse.getFailure();

                    boolean flag = Query(client,index,failure.getId());
                    if(!flag) {
                        log.info("Txid:" + failure.getMessage());
                    }
                }
            }
        }
    }

    public static SearchHits searchAfter(Object[] objects,
                                  int size,
                                  RestHighLevelClient client,
                                  String index) {
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(QueryBuilders.matchAllQuery());
        sourceBuilder.size(size);
        sourceBuilder.sort("_id", SortOrder.ASC);
        sourceBuilder.searchAfter(objects);

        SearchRequest searchRequest = new SearchRequest();
        searchRequest.indices(index);
        searchRequest.source(sourceBuilder);
        ActionFuture<SearchResponse> response = null;
        SearchResponse getResponse = null;
        try {
            getResponse = client.search(searchRequest,RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error("searchAfter error:" + e.getMessage());
        }

        SearchHits searchHits = getResponse.getHits();
        return searchHits;
    }

}
